package com.wxg.mq.ch02;

import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

public class SyncConsumer {
	public static void main(String[] args) throws MQClientException {
		DefaultMQPushConsumer consumer = 
				new DefaultMQPushConsumer(SyncProducer.GROUP_NAME);
		consumer.setNamesrvAddr(SyncProducer.NAMESRV_ADDR);
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//		consumer.setMessageModel(MessageModel.BROADCASTING);
		consumer.subscribe(SyncProducer.TOPIC, "*");
		consumer.registerMessageListener(new MessageListenerConcurrently() {
			@Override
			public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, 
					ConsumeConcurrentlyContext context) {
				System.out.printf(Thread.currentThread().getName() + 
						" Receive New Messages: " + msgs + "%n");
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			}
		});
		
		consumer.start();
	}
}
